package com.stqf.scala.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.io.RichOutputFormat;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.SortPartitionOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * Created By admin
 * </br></>Date: 2020/4/1
 * </br>Description:
 * </br>Note: 需要制定日志, 否则会被优先级高的替代: -Dlog4j.configurationFile=log4j2.xml
 */
public class WordCount {

    private static final Logger log = LoggerFactory.getLogger(WordCount.class);

    public static void main(String[] args) throws Exception {
        ParameterTool ptItem = ParameterTool.fromArgs(args);
        String filename = ptItem.get("file", "logs/full.0.log");
        // String filename = Paths.get(System.getProperty("user.dir"), "refer", "dbscript", "softly-1.0-cham.sql").toString();

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<String> dataSource = env.readTextFile(filename);

        SortPartitionOperator<Tuple2<String, Integer>> spo = dataSource.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (line, out) -> {
            String[] wordArr = line.split("[,\"<>(){}.\\s\\[\\]\\\\:;@]+");
            List<String> words = Stream.of(wordArr)
                    .filter(item -> item.matches("[a-zA-Z]+"))
                    .collect(Collectors.toList());
            log.debug("words:{}", words);
            words.forEach(word -> out.collect(new Tuple2<>(word, 1)));
        }).returns(Types.TUPLE(Types.STRING, Types.INT)).groupBy(0).sum(1).sortPartition(1, Order.DESCENDING);

        spo.output(new RichOutputFormat<Tuple2<String, Integer>>() {
            @Override
            public void configure(Configuration parameters) {
                log.info("{}", parameters);
            }

            @Override
            public void open(int taskNumber, int numTasks) throws IOException {
                log.info("{} ... {}", taskNumber, numTasks);
            }

            @Override
            public void writeRecord(Tuple2<String, Integer> record) throws IOException {
                log.info("[{} ... {}]", record.f0, record.f1);
            }

            @Override
            public void close() throws IOException {
                log.info("close ... ");
            }
        });

        // spo.print();
        env.execute();

	// System.out.println(java.util.Arrays.toString(args));

    }

}
